﻿using System;
using System.Threading;
using dotnet_etcd;
using Etcdserverpb;
using Fetches.Fetch.Factory.Persistence;
using FetchesDeploy.Distributed.DataPersistence;
using Grpc.Core;
using Mvccpb;
using Newtonsoft.Json;
using Serilog;

namespace FetchesDeploy.Distributed.UnitMonitor
{
    /*etcd目录结构设计
/fetches/config/workshop 平台配置
/fetches/FetchOrders/{id,FetchOrder} 任务存储，包含任务ID和FetchOrder
/fetches/ProcessOrders/{id,ProcessOrder} 任务存储，包含任务ID和ProcessOrder

/fetches/workshop/:id/ 容器注册对应ID并心跳保持/控制中心WatchRange监控变化以更新调度信息
/fetches/works/:id/ 容器自动更新自身运行的fetch\process任务列表并心跳保存/控制中心出现问题后根据该信息重构

/fetches/fetchWork/:id 控制中心写入要分配的爬取任务ID和FetcherId列表/容器获取该信息后删除该键
/fetches/processWork/:id 控制中心写入要分配的处理任务ID/容器获取该信息后删除该键

/fetches/waiting/fetch 待分配列表
/fetches/waiting/process
 */

    /* 环境变量ETCDCTL_API=3
       * watch包括子节点 etcdctl watch /fetches/task --prefix
       * 存数据 etcdctl put /fetches/task val
       * 取数据 etcdctl get /fetches/task          
       */

    public class EtcdUnitMonitor:IUnitMonitor
    {
        protected EtcdClient _etcdClient;
        private Action<UnitSchedules> _onlineAction;
        private Action<long> _offlineAction;

        public EtcdUnitMonitor(string etcdAddress=null)
        {
            try
            {
                if (string.IsNullOrEmpty(etcdAddress))
                {
                    etcdAddress = "http://localhost:2379";
                }
                _etcdClient = new EtcdClient(etcdAddress);
              
            }
            catch (RpcException e)
            {
                Log.Fatal("etcd未就绪,请启动etcd后尝试重启控制中心");
                throw;
            }

            //一个线程监控Workshop上下线
            new Thread(() =>
            {
                //dotnet_etcd中watch包括子节点的方法是WatchRange
                string key = EtcdRuntimeDataService.KeyForFactoryUnit;
                _etcdClient.WatchRange(key, WorkshopChanged);
            }).Start();

            //一个线程获得任务结束通知。每一个watch要单独的线程
            //new Thread(() =>
            //{
            //    var KeyForFinishedFetch = "/fetches/finished/fetch";
            //    _etcdClient.WatchRange(KeyForFinishedFetch, responses =>
            //    {
            //        foreach (var @event in responses)
            //        {
            //            if (@event.Type == Event.Types.EventType.Put)
            //            {
            //                var idStr = @event.Key.Replace($"{KeyForFinishedFetch}/", "");
            //                var taskId = Convert.ToInt64(idStr);
            //                cancelFetchAction(taskId);
            //                _etcdClient.Delete(@event.Key);
            //            }
            //        }
            //    });
            //}).Start();

            Log.Information("Factory is now watching etcd {address} for unit change", etcdAddress);
        }


        #region Workshop上下线工具方法

        private void WorkshopChanged(WatchResponse response)
        {
            foreach (var @event in response.Events)
            {
                try
                {
                    var idStr = @event.Kv.Key.ToStringUtf8().Replace(EtcdRuntimeDataService.KeyForFactoryUnit, "");
                    var workshopId = Convert.ToInt64(idStr);
                    if (@event.Type == Event.Types.EventType.Put)
                    {
                        var workshop = JsonConvert.DeserializeObject<UnitSchedules>(@event.Kv.Value.ToStringUtf8());
                        _onlineAction(workshop);
                    }
                    else
                    {
                        _offlineAction(workshopId);
                    }
                }
                catch (Exception e)
                {
                    Log.Error("an error occured when watching workshop change,key:{key},message:{message}",
                        @event.Kv.Key.ToStringUtf8(), e.Message);
                }
            }
        }

        #endregion

        public void Subscribe(Action<UnitSchedules> onlineAction, Action<long> offlineAction)
        {
            _onlineAction = onlineAction;
            _offlineAction = offlineAction;
        }
    }
}
